package com.lpssfxy.offline.utils

import com.lpssfxy.offline.entities.Product
import com.lpssfxy.offline.entities.ProductRating
import org.apache.spark.SparkConf
import org.apache.spark.sql.{Dataset, SparkSession}

/**
 * 工具类
 */
object AppUtils {

  val MONGODB_PRODUCT_COLLECTION = "product" // 商品集合
  val MONGODB_RATING_COLLECTION = "rating" // 定义要读取MongoDB中的表
  val MONGODB_USER_RECS_COLLECTION = "userRecs" // 定义根据ALS算法得到的用户推荐列表数据在Mongo中存放的表
  val MONGODB_RELATED_RECS_COLLECTION = "relatedProductSimList" // 用于存储商品及其相关性商品的集合
  val USER_MAX_RECOMMENDATION_SIZE = 20 // 定义最大只取出用户推荐列表中的前20个相似度的用户
  val MONGODB_PRODUCT_RECS_COLLECTION = "productRecs" // 定义根据ALS算法得到的商品相似度矩阵数据在Mongo中存放的表

  val configMap = Map(
    "spark.cores" -> "local[*]",
    "mongo.uri" -> "mongodb://fooadmin:123456@s3:27017/bigdata",
    "mongo.db" -> "bigdata"
  )

  def getSparkCores: String = configMap("spark.cores")
  def getMongoUri: String = configMap("mongo.uri")
  def getMongoDb: String = configMap("mongo.db")

  /**
   * 获取spark运行时setMaster对应的运行模式
   * @param args
   * @return
   */
  def getMaster(args: Array[String]): String = {
    // 检查是否传入了 master 参数
    if (args.length < 1) {
      println("请传入 master 参数，例如：local[*] 或 yarn 等")
      System.exit(1)
    }
    // 从命令行参数中获取 master 的值
    args(0)
  }

  /**
   * 创建SparkSession对象
   * @param config 配置信息
   * @return SparkSession对象
   */
  def createSparkSession(appName: String,master: String): SparkSession = {
    val sparkConf = new SparkConf().setAppName(appName)//.setMaster(master)
    SparkSession.builder().config(sparkConf).getOrCreate()
  }

  /**
   * 从MongoDB加载商品数据
   *
   * @param spark       SparkSession对象
   * @param mongoConfig MongoConfig对象
   * @return 商品数据的RDD
   */
  def loadProductData(spark: SparkSession): Dataset[Product] = {
    import spark.implicits._
    spark
      .read
      .option("uri", getMongoUri)
      .option("collection", MONGODB_PRODUCT_COLLECTION)
      .format("com.mongodb.spark.sql")
      .load()
      .as[Product]
      .cache()
  }

  /**
   * 从MongoDB加载评分数据
   *
   * @param spark       SparkSession对象
   * @param mongoConfig MongoConfig对象
   * @return 评分数据的RDD
   */
  def loadRatingData(spark: SparkSession) = {
    import spark.implicits._
    spark
      .read
      .option("uri", getMongoUri)
      .option("collection", MONGODB_RATING_COLLECTION)
      .format("com.mongodb.spark.sql")
      .load()
      .as[ProductRating]
      .rdd
      .map(rating => (rating.userId, rating.productId, rating.score))
      .cache()
  }

  /**
   * 将推荐结果保存到MongoDB
   *
   * @param recsDF 推荐列表的DataFrame
   */
  def saveRecommendationsToMongoDB(recsDF: org.apache.spark.sql.DataFrame, collection: String): Unit = {
    recsDF
      .write
      .option("uri", getMongoUri)
      .option("collection", collection)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()
  }

}
